package com.ls.rabbitmqService.direct;

import com.ls.rabbitmqService.topic.TopicSend;
import com.ls.rabbitmqService.utils.ConnectionFactoryUtil;
import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class DirectSend {

    private static final Logger logger = LoggerFactory.getLogger(TopicSend.class);



    public static void send(boolean confirm,boolean durable,String exchange_name,String queue_name,String routingKey,String msg){
        if(confirm){
            confirmSend(durable,exchange_name,queue_name,routingKey,msg);
        }else{
            send(durable,exchange_name,queue_name,routingKey,msg);
        }
    }

    private static void send(boolean durable,String exchange_name,String queue_name,String routingKey,String msg){
        Connection connection =null;
        Channel channel=null;
        try {
            connection=ConnectionFactoryUtil.getRabbitConnection();
            channel = connection.createChannel();
            channel.exchangeDeclare(exchange_name, BuiltinExchangeType.DIRECT,durable);
            //发送消息时设置消息的持久化
            AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
            //这里的deliveryMode=1代表不持久化，deliveryMode=2代表持久化。
            builder.deliveryMode(durable?2:1);
            AMQP.BasicProperties properties = builder.build();

            //queue队列的持久化 durable 参数代表是否需要持久化，true代表是  rabbitmq服务器重起可以保证消息可以再次被消费
            // 声明队列【参数说明：参数一：队列名称，参数二：是否持久化；参数三：是否独占模式；参数四：消费者断开连接时是否删除队列；参数五：消息其他参数】
            channel.queueDeclare(queue_name,durable,false,false,null);


            channel.basicPublish(exchange_name,routingKey,properties,msg.getBytes());
            logger.info(String.format("send routingKey:%s,message:%s",routingKey,msg));
        } catch (IOException e) {
            logger.error(e.getMessage(),e);
        } finally{
            if(channel!=null){
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
            if(connection!=null){
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

        }
    }


    /**
     * 开启发送方消息确认，能知道消息是否到达 broker 中间件
     * 可能消息在传递过来的时候因为特殊情况导致失败啦，这样可以记录日志知道
     * 消息根本进没有到交换器就失败啦
     */
    private static void confirmSend(boolean durable,String exchange_name,String queue_name,String routingKey,String msg){
        Connection connection =null;
        Channel channel=null;
        try {
            connection=ConnectionFactoryUtil.getRabbitConnection();
            channel = connection.createChannel();
            channel.exchangeDeclare(exchange_name, BuiltinExchangeType.DIRECT,durable);
            //发送消息时设置消息的持久化
            AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
            //这里的deliveryMode=1代表不持久化，deliveryMode=2代表持久化。
            builder.deliveryMode(durable?2:1);
            AMQP.BasicProperties properties = builder.build();

            //queue队列的持久化 durable 参数代表是否需要持久化，true代表是  rabbitmq服务器重起可以保证消息可以再次被消费
            // 声明队列【参数说明：参数一：队列名称，参数二：是否持久化；参数三：是否独占模式；参数四：消费者断开连接时是否删除队列；参数五：消息其他参数】
            channel.queueDeclare(queue_name,durable,false,false,null);

            channel.queueBind(queue_name,exchange_name,routingKey);

            channel.confirmSelect();
            channel.basicPublish(exchange_name,routingKey,properties,msg.getBytes());
            long nextPublishSeqNo = channel.getNextPublishSeqNo();
            logger.info("nextPublishSeqNo:"+nextPublishSeqNo);


            //1.普通发送方确认  每次发送1条消息确认1次 等待服务器端confirm 如果服务器返回false或者超时，客户端进行消息重传
           /* if (channel.waitForConfirms()) {
                System.out.println("消息发送成功" );
            }*/

            //2.批量Confirm模式  如果消息经常丢失这种方式效率不高，因为丢失会重发
            //channel.waitForConfirmsOrDie(); //直到所有信息都发布，只要有一个未确认就会IOException

            //3.异步监听确认和未确认的消息
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    /**
                     *暂时先不处理
                     *没用发送成功的消息暂时先不处理
                     */
                    logger.error(String.format("未确认消息，标识：%d，多个消息：%b", deliveryTag, multiple));
                }
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    logger.info("deliveryTag:"+deliveryTag);
                    //可以看到批量发送的消息并不是通过1次进行确认的有时候是多条一块确认的，可能是很多条
                    if(logger.isDebugEnabled())
                        logger.debug(String.format("已确认消息，标识：%d，多个消息：%b", deliveryTag, multiple));
                }
            });
            logger.info(String.format("send routingKey:%s,message:%s",routingKey,msg));
        } catch (IOException e) {
            logger.error(e.getMessage(),e);
        } finally{
            /*if(channel!=null){
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
            if(connection!=null){
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }*/

        }
    }



}
